#include <pthread.h>
#include <stdatomic.h>
#include <sys/prctl.h>

#include "utils.h"
#include "threadqueue.h"
#include "threadpool.h"

struct _threadtask_t
{
    void (*function)(void *arg);
    void *arg;
};

typedef struct _threaddata_t
{
    size_t index;
    pthread_t thread_id;
    struct timespec start_time;
    struct timespec stop_time;
    void *embedded_pool;
} threaddata_t;

struct _fix_threadpool_t
{
    threaddata_t *threads;
    threadboundedqueue_t *tasks;
    size_t thread_capacity;
    atomic_bool is_run;
};

int32_t fix_threadpool_destroy(fix_threadpool_t *pool)
{
    if (NULL == pool)
    {
        LOG_PRINT_ERROR("invalid param!");
        return THREADPOOL_RET_ERR_ARG;
    }

    if (!atomic_load(&pool->is_run))
    {
        LOG_PRINT_ERROR("already destroy!");
        return THREADPOOL_RET_ERR_AGAIN;
    }

    threadboundedqueue_stop(pool->tasks);
    atomic_store(&pool->is_run, false);

    for (size_t i = 0; i < pool->thread_capacity; ++i)
    {
        if (0 != pthread_join(pool->threads[i].thread_id, NULL))
        {
            LOG_PRINT_ERROR("pthread_join fail, errno[%d](%s)", errno, strerror(errno));
        }
        struct timespec now = time_mono();
        LOG_PRINT_INFO("thread[%ld]-join[%ld-%ld]", pool->threads[i].thread_id,
                       now.tv_sec, now.tv_nsec);
    }
    threadboundedqueue_uninit(&pool->tasks);
    free(pool->threads);
    free(pool);

    return THREADPOOL_RET_SUCCESS;
}

static void *fix_threadworker(void *args)
{
    int32_t ret = 0;
    threaddata_t *thread_data = (threaddata_t *)args;

    thread_data->start_time = time_mono();
    LOG_PRINT_INFO("thread[%ld]-start[%ld-%ld]", thread_data->thread_id,
                   thread_data->start_time.tv_sec, thread_data->start_time.tv_nsec);

    fix_threadpool_t *pool = (fix_threadpool_t *)thread_data->embedded_pool;
    char thread_name[16] = {};

    snprintf(thread_name, sizeof(thread_name), "tp_%zu", thread_data->index);
    prctl(PR_SET_NAME, thread_name, 0, 0, 0);

    while (atomic_load(&pool->is_run))
    {
        threadtask_t task = {};
        size_t task_size = sizeof(task);
        ret = threadboundedqueue_pop_block(pool->tasks, &task, &task_size, 0);
        if (THREADQUEUE_RET_SUCCESS == ret)
        {
            task.function(task.arg);
        }
        else if (THREADQUEUE_RET_ERR_STOP == ret)
        {
            break;
        }
        else
        {
            LOG_PRINT_ERROR("threadboundedqueue_pop_block fail, ret[%d]", ret);
            sleep(1);
            continue;
        }
    }

    thread_data->stop_time = time_mono();
    LOG_PRINT_INFO("thread[%lu]-stop[%ld-%ld]", (unsigned long)thread_data->thread_id,
                   thread_data->stop_time.tv_sec, thread_data->stop_time.tv_nsec);

    return NULL;
}

fix_threadpool_t *fix_threadpool_create(size_t thread_capacity, size_t queue_size)
{
    int32_t ret = 0;
    fix_threadpool_t *pool = NULL;
    size_t created_success_thread = 0;

    pool = (fix_threadpool_t *)calloc(1, sizeof(fix_threadpool_t));
    if (NULL == pool)
    {
        LOG_PRINT_ERROR("calloc fail, errno[%d](%s)!", errno, strerror(errno));
        return NULL;
    }

    pool->threads = (threaddata_t *)calloc(thread_capacity, sizeof(threaddata_t));
    if (NULL == pool->threads)
    {
        LOG_PRINT_ERROR("calloc fail, errno[%d](%s)!", errno, strerror(errno));
        free(pool);
        return NULL;
    }

    pool->tasks = threadboundedqueue_create(queue_size, sizeof(threadtask_t));
    if (NULL == pool->tasks)
    {
        LOG_PRINT_ERROR("threadboundedqueue_create fail, errno[%d](%s)!", errno, strerror(errno));
        free(pool->threads);
        free(pool);
        return NULL;
    }

    pool->thread_capacity = thread_capacity;
    atomic_store(&pool->is_run, true);

    for (size_t i = 0; i < thread_capacity; ++i)
    {
        pool->threads[i].index = i;
        pool->threads[i].embedded_pool = pool;
        ret = pthread_create(&pool->threads[i].thread_id, NULL, fix_threadworker, &pool->threads[i]);
        if (0 != ret)
        {
            LOG_PRINT_ERROR("pthread_create fail, ret[%d] errno[%d](%s)", ret, errno, strerror(errno));
            threadboundedqueue_uninit(&pool->tasks);
            atomic_store(&pool->is_run, false);

            for (size_t j = 0; j < created_success_thread; ++j)
            {
                if (0 != pthread_join(pool->threads[j].thread_id, NULL))
                {
                    LOG_PRINT_ERROR("pthread_join fail, errno[%d](%s)", errno, strerror(errno));
                }
            }

            free(pool->threads);
            free(pool);

            return NULL;
        }
        ++created_success_thread;
    }

    return pool;
}

int32_t fix_threadpool_addtask(fix_threadpool_t *pool, void (*function)(void *), void *arg)
{
    if (!pool || !function)
    {
        LOG_PRINT_ERROR("invalid param!");
        return THREADPOOL_RET_ERR_ARG;
    }

    if (!atomic_load(&pool->is_run))
    {
        LOG_PRINT_ERROR("already destroy!");
        return THREADPOOL_RET_ERR_AGAIN;
    }

    int32_t ret = 0;
    threadtask_t task = {};
    task.function = function;
    task.arg = arg;

    ret = threadboundedqueue_push_nonblock(pool->tasks, &task, sizeof(task));
    if (THREADQUEUE_RET_SUCCESS != ret)
    {
        LOG_PRINT_ERROR("threadboundedqueue_push_nonblock fail, ret[%d]", ret);
        if (THREADQUEUE_RET_ERR_FULL == ret)
        {
            ret = THREADPOOL_RET_ERR_QUEUE_FULL;
        }
        else
        {
            ret = THREADPOOL_RET_ERR_OTHER;
        }
    }
    else
    {
        ret = THREADPOOL_RET_SUCCESS;
    }

    return ret;
}

const char *fix_threadpool_strerror(int err)
{
    switch (err)
    {
#define X(code, name, msg)      \
    case THREADPOOL_RET_##name: \
        return msg;
        THREADPOOL_FOREACH_ERR(X)
#undef X
        default:
            return "Unknown thread queue error";
    }
}